package com.azh.server.tables;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import com.azh.server.datam.DataManager;
import com.azh.server.parser.statement.Begin;
import com.azh.server.parser.statement.Create;
import com.azh.server.parser.statement.Delete;
import com.azh.server.parser.statement.Insert;
import com.azh.server.parser.statement.Select;
import com.azh.server.parser.statement.Update;
import com.azh.server.util.Parser;
import com.azh.server.vm.VersionManager;

/**
 * @author zhaozhenhang <zhaozhenhang@kuaishou.com>
 * Created on 2022-05-01
 */
public class TableManagerImpl implements TableManager{

    public VersionManager vm;

    public DataManager dm;

    private Bootstarter starter;

    private Map<String, Table> tableCache;

    private Map<Long, List<Table>> txIdTableCache;
    private Lock lock;

    public TableManagerImpl(VersionManager vm, DataManager dm, Bootstarter starter) {
        this.vm = vm;
        this.dm = dm;
        this.starter = starter;
        this.tableCache = new HashMap<>();
        this.txIdTableCache = new HashMap<>();
        lock = new ReentrantLock();

    }

    private void loadTables() {
        long uid = firstTableUid();
        while (uid != 0) {
            Table table = Table.loadTable(this, uid);
        }
    }

    private long firstTableUid() {
        byte[] raw = starter.load();
        return Parser.parseLong(raw);
    }

    private void updateFirstTableUid(long uid) {
        byte[] raw = Parser.longToByte(uid);
        starter.update(raw);
    }

    @Override
    public BeginRes begin(Begin begin) {
        BeginRes res = new BeginRes();
        int level = begin.isRepeatableRead ? 1 : 0;
        res.txId = vm.begin(level);
        res.result = "begin".getBytes(StandardCharsets.UTF_8);
        return res;
    }

    @Override
    public byte[] commit(long txId) throws Exception {
        vm.commit(txId);
        return "commit".getBytes();
    }

    @Override
    public byte[] abort(long txId) {
        vm.abort(txId);
        return "abort".getBytes();
    }

    @Override
    public byte[] show(long txId) {
        lock.lock();
        try {
            StringBuilder builder = new StringBuilder();
            for (Table table : tableCache.values()) {
                builder.append(table.toString()).append("\n");
            }
            List<Table> tables = txIdTableCache.get(txId);
            if (tables == null) {
                return "\n".getBytes(StandardCharsets.UTF_8);
            }
            for (Table t : tables) {
                builder.append(t.toString()).append("\n");
            }
            return builder.toString().getBytes(StandardCharsets.UTF_8);
        }finally {
            lock.unlock();
        }
    }

    @Override
    public byte[] create(long txId, Create create) throws Exception {
        lock.lock();
        try {
            if(tableCache.containsKey(create.tableName)) {
                throw new RuntimeException("Duplicated table!");
            }
            Table table = Table.createTable(this, firstTableUid(), txId, create);
            updateFirstTableUid(table.uid);
            tableCache.put(create.tableName, table);
            if (!txIdTableCache.containsKey(txId)) {
                txIdTableCache.put(txId, new ArrayList<>());
            }
            txIdTableCache.get(txId).add(table);
            return ("create " + create.tableName).getBytes();
        } finally {
            lock.unlock();
        }
    }

    @Override
    public byte[] insert(long txId, Insert insert) throws Exception {
        lock.lock();
        Table table = tableCache.get(insert.tableName);
        lock.unlock();
        if(table == null) {
            throw new RuntimeException("Table not found!");
        }
        table.insert(txId, insert);
        return "insert".getBytes();
    }

    @Override
    public byte[] read(long txId, Select select) throws Exception {
        lock.lock();
        Table table = tableCache.get(select.tableName);
        lock.unlock();
        if(table == null) {
            throw new RuntimeException("Table not found!");
        }
        return table.read(txId, select).getBytes();
    }

    @Override
    public byte[] update(long txId, Update update) throws Exception {
        lock.lock();
        Table table = tableCache.get(update.tableName);
        lock.unlock();
        if(table == null) {
            throw new RuntimeException("Table not found!");
        }
        int count = table.update(txId, update);
        return ("update" + count).getBytes(StandardCharsets.UTF_8);
    }

    @Override
    public byte[] delete(long txId, Delete delete) throws Exception {
        lock.lock();
        Table table = tableCache.get(delete.tableName);
        lock.unlock();
        if(table == null) {
            throw new RuntimeException("Table not found!");
        }
        int count = table.delete(txId, delete);
        return ("delete " + count).getBytes();
    }
}
